package com.atguigu.pro3


import java.util

import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.{PatternSelectFunction, PatternTimeoutFunction}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * @description: individual patterns,个体模式可以包括单例模式和循环模式,单例模式只接收一个事件,而循环模式可以接收多个,可以在一个个体模式后追加两次,也就是指定循环次数
 * @time: 2021/4/6 14:30 订单超时事件
 * @author: baojinlong
 * */


// 定义输入输出样理类型
case class OrderEvent(orderId: Long, eventType: String, txId: String, timestamp: Long)

case class OrderResult(orderId: Long, resultMsg: String)

object CepDemo01 {
  def main(args: Array[String]): Unit = {
    /**
     * 匹配出现4次 start.times(4)     匹配出现2,3或者4次,并且尽可能多地重复 start.times(2,4).greedy
     * 匹配出现0或者4次 start.times(4).optional 匹配出现1次或多次start.oneOrMore
     * 匹配出现2,3或者4次 start.times(2,4) 匹配出现0次,2次或多次,并且尽可能多地重复 start.timesOrMore(2).optional.greedy
     * 个体模式的条件,每隔模式都需要指定触发条件,作为模式是否接受事件进入的判断依据
     * cep中的个体模式主要通过调用.where or until来指定条件
     * 严格近邻 strict contiguity 宽松relaxed 非确定性宽松紧邻non-deterministic relaxed contiguity
     * 匹配事件的提取
     */
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 不影响输出顺序
    environment.setParallelism(1)
    // 有事件时间的都设置这个时间
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 0.从文件中读取数据
    val orderEventStream: KeyedStream[OrderEvent, Long] = environment.readTextFile("src/main/resources/OrderLog.csv")
      .map(data => {
        val arr: Array[String] = data.split(",")
        OrderEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000)
      .keyBy(_.orderId)

    // 定义一个pattern
    val orderPayPattern: Pattern[OrderEvent, OrderEvent] = Pattern
      .begin[OrderEvent]("create").where(_.eventType == "create")
      // 宽松近邻
      .followedBy("pay").where(_.eventType == "pay")
      .within(Time.minutes(15))


    // 将pattern应用到数据流上,进行模式检测
    val patternStream: PatternStream[OrderEvent] = CEP.pattern(orderEventStream, orderPayPattern)
    // 定义侧输出流标签,用于处理超时事件
    val orderTimeoutOutputTag = new OutputTag[OrderResult]("orderTimeout")
    // 调用select方法,提取并处理匹配的成功支付事件以及超时事件
    val resultStream: DataStream[OrderResult] = patternStream
      .select(orderTimeoutOutputTag,
        new OrderTimeoutSelect,
        new OrderPaySelect
      )

    resultStream.print("payed")
    resultStream.getSideOutput(orderTimeoutOutputTag).print("orderTimeoutOutputTag")
    environment.execute("order timeout job")
  }

}


// 实现自定义的PatternTimeoutFunction以及PatternSelectFunction
class OrderTimeoutSelect extends PatternTimeoutFunction[OrderEvent, OrderResult] {
  override def timeout(pattern: util.Map[String, util.List[OrderEvent]], timeoutTimestamp: Long): OrderResult = {
    val timeoutOrderId: Long = pattern.get("create").iterator.next.orderId
    OrderResult(timeoutOrderId, "timeout:" + timeoutTimestamp)
  }
}

class OrderPaySelect extends PatternSelectFunction[OrderEvent, OrderResult] {
  override def select(pattern: util.Map[String, util.List[OrderEvent]]): OrderResult = {
    val payedOrderId: Long = pattern.get("pay").iterator.next.orderId
    OrderResult(payedOrderId, "payed successfully")
  }
}